package utils;

import common.Constant;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.twill.kafka.client.KafkaConsumer;

import java.util.Properties;

public class FlinkSourceUtil {
    public static SourceFunction<String> getKafkaSource(String topic, String groupId) {

        Properties props = new Properties();
        props.setProperty("group.id", groupId);
        props.setProperty("bootstrap.servers", Constant.KAFKA_BROKERS);
        FlinkKafkaConsumer<String> source = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);

        source.setStartFromLatest();
        return source;
    }
}
